package com.gitee.gateway.core;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonToken;
import com.fasterxml.jackson.databind.*;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.gitee.gateway.core.db.DatabaseManager;
import com.gitee.gateway.core.utils.Configuration;
import com.gitee.gateway.core.utils.StreamingParser;
import com.hazelcast.config.Config;
import io.vertx.core.DeploymentOptions;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.core.json.jackson.DatabindCodec;
import io.vertx.core.spi.cluster.ClusterManager;
import io.vertx.micrometer.MicrometerMetricsOptions;
import io.vertx.micrometer.VertxPrometheusOptions;
import io.vertx.spi.cluster.hazelcast.HazelcastClusterManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * gateway entry
 */
public class Server {

    private final static Logger LOGGER = LoggerFactory.getLogger(Server.class);

    static {
        configureJackson();
    }

    public static void main(String[] args) {
        DatabaseManager.testConnection();
        JsonObject config = Configuration.load();
        start(config);
    }

    /**
     * 集群模式启动
     *
     * @param serverConfig 配置
     */
    private static void start(JsonObject serverConfig) {
        JsonObject globalConfig = serverConfig.getJsonObject("global");
        VertxOptions vOptions = new VertxOptions(globalConfig);
        vOptions.setMaxEventLoopExecuteTimeUnit(TimeUnit.MILLISECONDS);
        vOptions.setMaxWorkerExecuteTimeUnit(TimeUnit.MILLISECONDS);
        //启用普罗米修斯插件
        JsonObject prometheusOptions = serverConfig.getJsonObject("prometheus");
        if (prometheusOptions != null && prometheusOptions.getBoolean("enabled", false)) {
            int port = prometheusOptions.getInteger("port", 8081);
            String path = prometheusOptions.getString("path", "/metrics");
            vOptions.setMetricsOptions(new MicrometerMetricsOptions()
                    .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true).setStartEmbeddedServer(true)
                            .setEmbeddedServerOptions(new HttpServerOptions().setPort(port))
                            .setEmbeddedServerEndpoint(path))
                    .setEnabled(true));
        }
        JsonObject clusterConfig = serverConfig.getJsonObject("cluster");
        if (clusterConfig != null && clusterConfig.getBoolean("enabled", false)) {
            //Hazelcast 默认使用 JUL，改为 slf4j
            System.setProperty("hazelcast.logging.type", "slf4j");
            //因为 tika 引入了 xerces 包导致 hazelcast 启动失败，所以通过环境变量来忽略这个错误
            boolean ignoreXxe = clusterConfig.getBoolean("ignore-xxe-protection-failures", false);
            System.setProperty("hazelcast.ignoreXxeProtectionFailures", String.valueOf(ignoreXxe));
            //设置心跳时间，超出此时间节点离线
            int noHeartBeat = clusterConfig.getInteger("max-no-heartbeat-seconds", 30);
            System.setProperty("hazelcast.max.no.heartbeat.seconds", String.valueOf(noHeartBeat));

            String configFile = "config" + File.separator + "hazelcast.yaml";
            try {
                Config hazelcastConfig = Config.loadFromFile(new File(configFile));
                ClusterManager mgr = new HazelcastClusterManager(hazelcastConfig);
                vOptions.setClusterManager(mgr);
            } catch (IOException e) {
                LOGGER.error("Failed to load " + configFile, e);
                System.exit(-1);
            }

            Vertx.clusteredVertx(vOptions)
                    .onSuccess(vertx -> startVertx(vertx, serverConfig))
                    .onFailure(e -> {
                        LOGGER.error("Failed to start in cluster.", e);
                        System.exit(-1);
                    });
        } else
            startVertx(Vertx.vertx(vOptions), serverConfig);
    }

    /**
     * 启动服务
     *
     * @param vertx  vertx 服务实例
     * @param config 全局配置信息
     */
    private static void startVertx(Vertx vertx, JsonObject config) {
        long startTime = System.currentTimeMillis();
        JsonObject services = Configuration.loadServices();
        List<Future<?>> futures = new ArrayList<>();
        services.stream().forEachOrdered(v -> {
            String serviceName = v.getKey();
            JsonObject serviceConfig = (JsonObject) v.getValue();
            if (!serviceConfig.getBoolean("enabled", true))
                return;
            try {
                String service = serviceConfig.getString("className");
                DeploymentOptions options = new DeploymentOptions(serviceConfig);
                futures.add(vertx.deployVerticle(service, options)
                        .onSuccess(n -> LOGGER.info("service: {} -> {} ({}) started.", serviceName, service, options.getInstances()))
                        .onFailure(res -> {
                            LOGGER.error("Failed to deploy service[" + serviceName + "]", res);
                            System.exit(-1);
                        }));
            } catch (Exception e) {
                LOGGER.error("Failed to start service " + serviceName, e);
                System.exit(-1);
            }
        });
        Future.all(futures).onSuccess(v -> {
            //install shutdown hook
            Runtime.getRuntime().addShutdownHook(new Thread(() -> stop(vertx)));
            LOGGER.info("started in {} ms.", (System.currentTimeMillis() - startTime));
        }).onFailure(e -> {
            LOGGER.error("Failed to start services.", e);
            System.exit(-1);
        });
    }

    private static void stop(Vertx vertx) {
        LOGGER.info("Vertx stopping...");
        vertx.close(res -> {
            if (res.succeeded())
                LOGGER.info("Vertx stopped.");
            else
                LOGGER.error("Failed to stop vertx.", res.cause());
        });
    }

    /**
     * 对 Jackson 进行一些初始配置
     */
    private static void configureJackson() {
        ObjectMapper mapper = DatabindCodec.mapper();
        mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        mapper.setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
        mapper.registerModule(new SimpleModule()
                .addDeserializer(JsonObject.class, new JsonDeserializer<>() {
                    @Override
                    public JsonObject deserialize(JsonParser parser, DeserializationContext dc) throws IOException {
                        if (parser.currentToken() == JsonToken.VALUE_STRING)
                            return new JsonObject(parser.getText());
                        return StreamingParser.readJsonObject(parser);
                    }
                })
                .addDeserializer(JsonArray.class, new JsonDeserializer<>() {
                    @Override
                    public JsonArray deserialize(JsonParser parser, DeserializationContext dc) throws IOException {
                        if (parser.currentToken() == JsonToken.VALUE_STRING)
                            return new JsonArray(parser.getText());
                        return StreamingParser.readJsonArray(parser);
                    }
                }));
    }

}
